package top.zhaoyuliu.logbook.client.connect.client;

import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock;
import java.util.stream.Collectors;

import org.tio.client.ClientChannelContext;
import org.tio.client.ClientGroupContext;
import org.tio.client.TioClient;
import org.tio.client.intf.ClientAioHandler;
import org.tio.core.ChannelContext;
import org.tio.core.Node;
import org.tio.core.Tio;
import org.tio.core.intf.Packet;
import org.tio.core.stat.ChannelStat;
import org.tio.utils.SystemTimer;
import org.tio.utils.lock.SetWithLock;

import cn.hutool.core.collection.CollectionUtil;
import lombok.extern.slf4j.Slf4j;
import top.zhaoyuliu.logbook.client.compent.AppenderCompent;
import top.zhaoyuliu.logbook.client.constant.Const;
import top.zhaoyuliu.logbook.client.properties.LogbookClientProperties;
import top.zhaoyuliu.logbook.client.queue.LbQueue;

/**
 * Copyright © 2021 Arain. All rights reserved.
 *
 * @Title: ClientServerStart.java
 * @Prject: logbook-client
 * @Package: top.zhaoyuliu.logbook.client.connect
 * @Description: <功能详细描述>
 * @author: Arain
 * @date: 2021年1月22日 下午2:52:01
 * @version: V1.0
 */
@Slf4j
public class ClientServerStart {

	private ClientGroupContext clientGroupContext;

	private LogbookClientProperties logbookClientProperties;

	public ClientServerStart(ClientGroupContext clientGroupContext, LogbookClientProperties logbookClientProperties) {
		this.clientGroupContext = clientGroupContext;
		this.logbookClientProperties = logbookClientProperties;
	}

	public void start() {
		log.info("客户端初始化");
		LbQueue.pushSize = logbookClientProperties.getLogNumPush();
		try {
			List<String> cluster = logbookClientProperties.getCluster();
			if (CollectionUtil.isNotEmpty(cluster)) {
				Set<Node> serverNodes = new HashSet<>();
				for (String address : cluster) {
					String[] add = address.split(":");
					if (add.length != 2) {
						continue;
					}
					try {
						Node node = new Node(add[0], Integer.parseInt(add[1]));
						serverNodes.add(node);
					} catch (Exception e) {
						log.warn("连接失败 -> {}", address);
						continue;
					}
				}
				bindServerGroup(serverNodes);
			} else {
				Node node = new Node(logbookClientProperties.getHost(), logbookClientProperties.getPort());
				TioClient client = new TioClient(clientGroupContext);
				ClientChannelContext clientChannelContext = client.connect(node);
				startHeartbeatTask();
				AppenderCompent.setClientChannelContext(clientChannelContext);
			}
			AppenderCompent.setAppId(logbookClientProperties.getAppId());
			AppenderCompent.addExcludePackages(Const.EXCLUDEPACKAGE.split(","));
		} catch (Exception e) {
			e.printStackTrace();
			log.error("客户端初始化失败");
		}

	}

	private void bindServerGroup(Set<Node> serverNodes) {
		SetWithLock<ChannelContext> setWithLock = Tio.getAllChannelContexts(clientGroupContext);
		Lock lock2 = setWithLock.getLock().readLock();
		lock2.lock();
		try {
			Set<ChannelContext> set = setWithLock.getObj();
			Set<Node> connectedNodes = set.stream().map(ChannelContext::getServerNode).collect(Collectors.toSet());
			for (Node node : serverNodes) {
				if (!connectedNodes.contains(node)) {
					try {
						TioClient client = new TioClient(clientGroupContext);
						ClientChannelContext channelContext = client.connect(node);
						AppenderCompent.setClientChannelContext(channelContext);
					} catch (Exception e) {
						e.printStackTrace();
					}
				}
			}
			AppenderCompent.setClientGroupContext(clientGroupContext, clientGroupContext.connecteds.getObj());
			for (ChannelContext channelContext : set) {
				Node node = channelContext.getServerNode();
				if (!serverNodes.contains(node)) {
					Tio.remove(channelContext, "主动关闭" + node.getIp());
				}

			}
			startHeartbeatTask();
			Const.CLUSTER = true;
		} finally {
			lock2.unlock();
		}
	}

	private void startHeartbeatTask() {
		final ClientAioHandler aioHandler = clientGroupContext.getClientAioHandler();
		final String id = clientGroupContext.getId();
		new Thread(new Runnable() {
			@Override
			public void run() {
				while (!clientGroupContext.isStopped()) {
					SetWithLock<ChannelContext> setWithLock = clientGroupContext.connecteds;
					ReadLock readLock = setWithLock.readLock();
					readLock.lock();
					try {
						Set<ChannelContext> set = setWithLock.getObj();
						long currtime = SystemTimer.currTime;
						for (ChannelContext entry : set) {
							ClientChannelContext channelContext = (ClientChannelContext) entry;
							if (channelContext.isClosed || channelContext.isRemoved) {
								continue;
							}
							if (null == AppenderCompent.getClientChannelContext()) {
								AppenderCompent.setClientChannelContext(channelContext);
							}
							ChannelStat stat = channelContext.stat;
							long compareTime = Math.max(stat.latestTimeOfReceivedByte, stat.latestTimeOfSentPacket);
							long interval = currtime - compareTime;
							if (interval >= logbookClientProperties.getTimeout() / 2) {
								Packet packet = aioHandler.heartbeatPacket(channelContext);
								if (packet != null) {
									Tio.send(channelContext, packet);
								}
							}
						}
					} catch (Throwable e) {
						log.error("", e);
					} finally {
						try {
							readLock.unlock();
							Thread.sleep(logbookClientProperties.getTimeout() / 4);
						} catch (Throwable e) {
							log.error(e.toString(), e);
						} finally {

						}
					}
				}
			}
		}, "tio-timer-heartbeat" + id).start();
	}
}
